package com.actor;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import com.message.TransformationMessages;

import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.actor.Props;
import akka.actor.Terminated;
import akka.actor.UntypedActor;
import akka.cluster.Cluster;
import akka.cluster.ClusterEvent.CurrentClusterState;
import akka.cluster.Member;
import akka.pattern.Patterns;
import akka.util.Timeout;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;

public class ClusterClientMasterActor extends UntypedActor {

	private static ClusterClientMasterActor INSTANCE;

	public ClusterClientMasterActor() {
		INSTANCE = this;
	}

	public static ClusterClientMasterActor getInstance() {
		return INSTANCE;
	}

	List<ActorRef> backends = new ArrayList<ActorRef>();
	int jobCounter = 0;

	public Future<Object> askWorld(TransformationMessages.WorldPacket packet) {
		Future<Object> future = Patterns.ask(getSelf(), packet, new Timeout(FiniteDuration.create(5, TimeUnit.SECONDS)));
		return future;
	}

	public Future<Object> askGameServer(TransformationMessages.GameServerPacket resPacket) {
		CurrentClusterState state = Cluster.get(getContext().system()).state();
		for (Member member : state.getMembers()) {
			if (member.hasRole(String.valueOf(resPacket.getGameServerId()))) {
				ActorSelection actorSelection = getContext()
						.actorSelection(member.address() + "/user/myAkkaClusterClient");
				resPacket.setSameServer(true);
				Future<Object> future = Patterns.ask(actorSelection, resPacket,
						new Timeout(FiniteDuration.create(5, TimeUnit.SECONDS)));
				return future;
			}
		}
		return null;
	}
	
	public void sendToWrold(TransformationMessages.WorldPacket packet){
		jobCounter++;
		backends.get(jobCounter % backends.size()).forward(packet, getContext());
	}

	public void sendToGameServer(TransformationMessages.GameServerPacket resPacket) {
		CurrentClusterState state = Cluster.get(getContext().system()).state();
		for (Member member : state.getMembers()) {
			if (member.hasRole(String.valueOf(resPacket.getGameServerId()))) {
				ActorSelection actorSelection = getContext()
						.actorSelection(member.address() + "/user/myAkkaClusterClient");
				resPacket.setSameServer(false);
				actorSelection.tell(resPacket, getSelf());
				break;
			}
		}
	}

	@Override
	public void onReceive(Object message) {
		if ((message instanceof TransformationMessages.WorldPacket) && backends.isEmpty()) {// 无服务提供者
			// TransformationMessages.SNetPacket job =
			// (TransformationMessages.SNetPacket) message;
			// getSender().tell(new TransformationMessages.JobFailed("Service
			// unavailable, try again later"),
			// getSender());

		} else if (message instanceof TransformationMessages.WorldPacket) {
			TransformationMessages.WorldPacket job = (TransformationMessages.WorldPacket) message;
			if (job.isSameServer()) {
				/**
				 * 这里在客户端业务代码里进行负载均衡操作。实际业务中可以提供多种负载均衡策略，并且也可以做分流限流等各种控制。
				 */
				job.setSameServer(false);
				jobCounter++;
				backends.get(jobCounter % backends.size()).forward(job, getContext());
			} else {
				ActorRef ref = getContext().actorOf(Props.create(ClusterClientWorkerActor.class));
				ref.tell(message, getSender());
			}

		} else if (message instanceof TransformationMessages.GameServerPacket) {
			TransformationMessages.GameServerPacket packet = (TransformationMessages.GameServerPacket) message;
			if (packet.isSameServer()) {
				packet.setSameServer(false);
				getSender().tell(packet, this.getSelf());
			} else {
				ActorRef ref = getContext().actorOf(Props.create(ClusterClientWorkerActor.class));
				ref.tell(message, getSender());
			}
		} else if (message instanceof Integer) {
			int msg = (Integer) message;
			if (msg == TransformationMessages.BACKEND_REGISTRATION) {
				/**
				 * 注册服务提供者
				 */
				getContext().watch(getSender());// 这里对服务提供者进行watch
				backends.add(getSender());
			}

		} else if (message instanceof Terminated) {
			/**
			 * 移除服务提供者
			 */
			Terminated terminated = (Terminated) message;
			backends.remove(terminated.getActor());

		} else {
			unhandled(message);
		}
	}

}
